package cas.ihep.hdfs;

import com.google.common.io.Closer;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.hdfs.protocol.LocatedBlocks;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class HdfsFile implements Closeable,Iterable<HdfsBlock> {

    HdfsSystem mFs;
    String mFilepath;
    HdfsFileStatus mMetadata;
    //LocatedBlocks mBlocks;
    Closer mCloser;
    List<HdfsBlock> mBlocks;
    boolean mClosed;

    public HdfsFile(HdfsSystem fs,String path)throws IOException {
        mFs=fs;
        mFilepath=path;
        mMetadata=mFs.mClient.getFileInfo(path);
        LocatedBlocks blocks=mFs.mClient.getLocatedBlocks(path,0,mMetadata.getLen());
        mBlocks=new ArrayList<>(blocks.getLocatedBlocks().size());
        mCloser= Closer.create();
        for (LocatedBlock blk:blocks.getLocatedBlocks()) {
            mBlocks.add(mCloser.register(new HdfsReadableBlock(this, blk)));
        }
        mClosed=false;
    }

    public HdfsFile(HdfsSystem fs, Path path)throws IOException {
        this(fs,path.toUri().getPath());
    }

    public HdfsBlock getBlock(int idx)throws IOException {
        return mBlocks.get(idx);
    }

    public List<HdfsBlock> blocks(){
        return mBlocks;
    }

    public void close()throws IOException {
        if (!mClosed){
            mClosed=true;
            mCloser.close();
        }
    }

    @Override
    public Iterator<HdfsBlock> iterator() {
        return mBlocks.iterator();
    }
}